package abacab.remote.netty;

import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import abacab.remote.Connection;
import abacab.remote.ConnectionRegistry;

/**
 * The Netty implementation of a client connection
 */
@SuppressWarnings("synthetic-access")
public class NettyClient extends Connection {
	/** The logger */
	private static final Logger		logger	= LoggerFactory.getLogger(NettyServerConnection.class);

	/** The factory for channels */
	private final ChannelFactory	channelFactory;

	/** The host & port */
	private final String					hostPort;

	/** The handler for the channel */
	private final ChannelHandler	handler;

	/**
	 * Create a client connection
	 * 
	 * @param registry
	 *          the registry
	 * @param host
	 *          the host
	 * @param port
	 *          the port
	 * @throws ConnectException
	 *           if the connection could not be established in time
	 */
	public NettyClient(final ConnectionRegistry registry, String host, int port) throws ConnectException {
		super(registry);
		this.channelFactory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),
				Executors.newCachedThreadPool());
		this.hostPort = host + ":" + port;
		this.handler = new ChannelHandler();

		ClientBootstrap bootstrap = new ClientBootstrap(this.channelFactory);
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			@Override
			public ChannelPipeline getPipeline() {
				ChannelPipeline pipeline = Channels.pipeline();
				pipeline.addLast("frame", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 2, 0, 2));
				pipeline.addLast("connection", NettyClient.this.handler);
				return pipeline;
			}
		});
		bootstrap.setOption("tcpNoDelay", Boolean.TRUE);
		bootstrap.setOption("keepAlive", Boolean.TRUE);

		ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
		try {
			if (future.await().isSuccess()) {
				this.handler.setChannel(future.getChannel());
				register();
			}
			else {
				throw new ConnectException("Failed to connect");
			}
		}
		catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			throw new ConnectException("The thread was interrupted while connecting");
		}
	}

	@Override
	public String getRemoteHostPort() {
		return this.hostPort;
	}

	@Override
	public InetSocketAddress getRemoteAddress() {
		return (InetSocketAddress) this.handler.channel.getRemoteAddress();
	}

	@Override
	public void close() {
		this.handler.close();
		this.channelFactory.releaseExternalResources();
	}

	@Override
	protected void send(byte[] message) {
		this.handler.send(message);
	}

	/**
	 * The channel handler
	 */
	private final class ChannelHandler extends SimpleChannelHandler {
		/** The channel */
		private volatile Channel	channel;

		/**
		 * Set the channel
		 * 
		 * @param channel
		 *          the channel
		 */
		public void setChannel(Channel channel) {
			this.channel = channel;
		}

		/**
		 * Close the connection
		 */
		void close() {
			unregister();
			ChannelFuture future = this.channel.close();
			if (!future.awaitUninterruptibly(5000L)) {
				logger.warn("Failed to close the client connection in time (it may close in the future)");
			}
		}

		@Override
		public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
			unregister();
			super.channelDisconnected(ctx, e);
		}

		/**
		 * Send a message
		 * 
		 * @param message
		 *          the message
		 */
		protected void send(byte[] message) {
			ChannelBuffer buffer = ChannelBuffers.buffer(message.length + 2);
			buffer.writeShort(message.length);
			buffer.writeBytes(message);
			this.channel.write(buffer);
		}

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
			NettyClient.this.messageReceived(((ChannelBuffer) e.getMessage()).array());
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
			e.getCause().printStackTrace();
			e.getChannel().close();
		}
	}
}